package org.example.io;

import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.buffer.UnpooledHeapByteBuf;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Reactor模式
 */
@Slf4j
public class NIOServer {
    public static void main(String[] args) throws Exception{
        // 服务端启动
        // 有一个acceptor线程，负责接收Socket
        // 有一个worker线程
        Selector acceptor = Selector.open();
        Selector worker = Selector.open();

        ServerSocketChannel serverSocket = ServerSocketChannel.open();
        // 配置成NIO
        serverSocket.configureBlocking(false);
        serverSocket.socket().bind(new InetSocketAddress(8000));

        serverSocket.register(acceptor, SelectionKey.OP_ACCEPT);

        ExecutorService executorService = Executors.newFixedThreadPool(10);

        // 接收链接
        new Thread(() -> {
            try {
                while (true) {
                    if (acceptor.select(1) > 0) {
                        Set<SelectionKey> selectionKeys = acceptor.selectedKeys();
                        Iterator<SelectionKey> iterator = selectionKeys.iterator();
                        while (iterator.hasNext()) {
                            try {
                                SelectionKey itemKey = iterator.next();
                                SocketChannel socketChannel = ((ServerSocketChannel) itemKey.channel()).accept();
                                socketChannel.configureBlocking(false);
                                socketChannel.register(worker, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
                            } catch (IOException e) {
                                e.printStackTrace();
                            } finally {
                                iterator.remove();
                            }
                        }
                    }
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }).start();

        // worker线程
        new Thread(() -> {
            try {
                while (true) {
                    if (worker.select(1) > 0) {
                        Set<SelectionKey> selectionKeys = worker.selectedKeys();
                        // 每次有读的事件，都要开线程，显然不合理
//                        selectionKeys.forEach((item) -> {
//                            SocketChannel socketChannel = (SocketChannel) item.channel();
//                            // 实际业务处理的线程
//                            executorService.submit(()->{
//                                ByteBuffer buffer = ByteBuffer.allocate(1024);
//                                int len = 0;
//                                try {
//                                    while ((len = socketChannel.read(buffer)) > 0) {
//                                        log.info(new String(buffer.array(), 0, len));
//                                    }
//                                } catch (IOException e) {
//                                    e.printStackTrace();
//                                }
//                            });
//                        });

                        Iterator<SelectionKey> iterator = selectionKeys.iterator();
                        while (iterator.hasNext()) {
                            SelectionKey selectionKey = iterator.next();
                            if (selectionKey.isReadable()) {
                                SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                                ByteBuffer buffer = ByteBuffer.allocate(1024);
                                int len = socketChannel.read(buffer);
                                buffer.flip();
                                log.info("收到消息：" + new String(buffer.array(), 0, len));
                            }

                            iterator.remove();
                        }
                    }
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }).start();

        Thread.sleep(Integer.MAX_VALUE);
    }
}
